package main

import (
	"flag"
	"fmt"
	"log"
	"time"

	"github.com/nsqio/go-nsq"
)

func main() {
	go startConsumer()
	startProducer()
}

var url string
var url1 string

func init() {
	//具体ip,端口根据实际情况传入或者修改默认配置
	flag.StringVar(&url, "url", "127.0.0.1:4150", "nsqd")         //tcp
	flag.StringVar(&url1, "url1", "127.0.0.1:4161", "nsqlookupd") //http
	flag.Parse()
}

// 生产者
func startProducer() {
	cfg := nsq.NewConfig()
	producer, err := nsq.NewProducer(url, cfg)
	if err != nil {
		log.Fatal(err)
	}
	// 发布消息

	if err := producer.Publish("test", []byte("test message0")); err != nil {
		log.Fatal("publish error: " + err.Error())
	}
	if err := producer.Publish("test", []byte("test message1")); err != nil {
		log.Fatal("publish error: " + err.Error())
	}
	if err := producer.Publish("test", []byte("test message2")); err != nil {
		log.Fatal("publish error: " + err.Error())
	}
	if err := producer.Publish("test", []byte("test message3")); err != nil {
		log.Fatal("publish error: " + err.Error())
	}
	time.Sleep(1 * time.Second)

}

type nsqHandler struct {
	nsqConsumer      *nsq.Consumer
	messagesReceived int
}

//处理消息
func (nh *nsqHandler) HandleMessage(msg *nsq.Message) error {
	nh.messagesReceived++
	fmt.Println("nh.messagesReceived++ ======== ", nh.messagesReceived)
	fmt.Printf("receive ID:%s,addr:%s,message:%s", msg.ID, msg.NSQDAddress, string(msg.Body))
	fmt.Println()
	return nil
}

// 消费者
func startConsumer() {
	cfg := nsq.NewConfig()
	consumer, err := nsq.NewConsumer("test", "sensor01", cfg)
	if err != nil {
		log.Fatal(err)
	}

	// 设置消息处理函数：方式一
	/*

		consumer.AddHandler(
		nsq.HandlerFunc(func(message *nsq.Message) error {
			log.Println(string(message.Body))
			return nil
		}))

	*/
	//设置消息处理函数：方式二
	handler := func(message *nsq.Message) error {
		fmt.Println(string(message.Body))
		return nil
	}
	consumer.AddHandler(nsq.HandlerFunc(handler))

	//设置消息处理函数：方式三
	/*
		handler := &nsqHandler{nsqConsumer: consumer}
		consumer.AddHandler(handler)
	*/

	// nsqlookupd
	//[]string
	if err := consumer.ConnectToNSQLookupds([]string{url1}); err != nil {
		log.Fatal(err)
	}
	<-consumer.StopChan
}
